package studio.raptor.dispatcher.server.filter;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import java.beans.BeanInfo;
import java.beans.Introspector;
import java.beans.PropertyDescriptor;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import studio.raptor.databus.command.EventMetaData;
import studio.raptor.dispatcher.server.dao.DispatcherConfigDAO;
import studio.raptor.dispatcher.server.entity.BaseSubscriptionConfig;
import studio.raptor.dispatcher.server.entity.SubItem;

/**
 * 订阅关系
 *
 * @author jack
 * @author liuyong
 * @since 1.0
 */
public abstract class Subscription {

  private static final Logger log = LoggerFactory.getLogger(Subscription.class);


  private List<BaseSubscriptionConfig> baseSubscriptionConfigs;
  private Map<String/*code+name*/, String/*handleClass*/> otherSubConfigMap;
  private Set<String> itemNames;
  private static final Integer SEND = 0;
  private static final Integer NOT_SEND = 1;
  private String topic;

  /**
   * 订阅关系缓存，加速订阅关系判断 key：eventId，value：订阅当前事件的接口集合
   */
  private final Cache<String, Set<String>> subscribeTable = CacheBuilder.newBuilder()
      .maximumSize(10000).build();

  /**
   * 获取订阅当前事件的接口集合
   *
   * @param meta 事件的元信息
   * @return 订阅当前事件的接口集合
   */
  public Set<String> getSubscription(final EventMetaData meta) {
    Set<String> allowSendChannels = new HashSet<String>();
    final String businessNo = String.valueOf(meta.getBusinessNo());//object_key
    log.debug("filter the event:{}", meta.toString());

    try {
      allowSendChannels = subscribeTable.get(businessNo, new Callable<Set<String>>() {
        @Override
        public Set<String> call() throws Exception {
          Set<String> codeTable = new HashSet<String>();
          Set<String> codeTableNew = subscribeTable.getIfPresent(businessNo);
          if (codeTableNew != null) {
            return codeTableNew;
          } else {
            //base
            for (BaseSubscriptionConfig baseSubscriptionConfig : baseSubscriptionConfigs) {
              if (!isBaseSubscribed(baseSubscriptionConfig, meta)) {
                continue;
              }
              codeTable.add(baseSubscriptionConfig.getCodeName()
                  .substring(0, baseSubscriptionConfig.getCodeName().indexOf("-")));
            }
            //other
            for (Map.Entry<String, String> entry : otherSubConfigMap.entrySet()) {
              if (!isOtherSubscrpted(entry, convertObject2Map(meta))) {
                continue;
              }
              codeTable.add(entry.getKey().substring(0, entry.getKey().indexOf("-")));
              log.info("subscription:{} added,",
                  entry.getKey().substring(0, entry.getKey().indexOf("-")));
            }
            return codeTable;
          }
        }
      });
    } catch (Exception e) {
      log.error("get the {} subscription  error ", topic, e);
    }

    return allowSendChannels;
  }

  /**
   * 加载停复机 or 普通业务的表配置
   */
  public void loadSubConfig(String topic, DispatcherConfigDAO dispatcherConfigDAO) {
    Preconditions.checkArgument(!Strings.isNullOrEmpty(topic), "topic is null or empty!");

    this.topic = topic;

    itemNames = dispatcherConfigDAO.queryItemNames();
    otherSubConfigMap = dispatcherConfigDAO.queryOtherSubConfigByTopic(topic);
    baseSubscriptionConfigs = dispatcherConfigDAO.queryBaseSubConfigByTopic(topic);
  }

  /**
   * 判断订阅关系
   *
   * @param baseSubscriptionConfig 常规配置
   * @param meta 元信息
   * @return true or false
   */
  private boolean isBaseSubscribed(BaseSubscriptionConfig baseSubscriptionConfig,
      EventMetaData meta) throws Exception {
    boolean isSubscribed = false;
    Map<String, Object> itemMap = convertObject2Map(meta);
    if (SEND == baseSubscriptionConfig.getIsSend()) {
      isSubscribed = checkEachConfigSubscribed(baseSubscriptionConfig, itemMap);
    } else if (NOT_SEND == baseSubscriptionConfig.getIsSend()) {
      isSubscribed = !checkEachConfigSubscribed(baseSubscriptionConfig, itemMap);
    }
    return isSubscribed;
  }

  private boolean checkEachConfigSubscribed(BaseSubscriptionConfig baseSubscriptionConfig,
      Map<String, Object> itemMap) {
    boolean isSubscribed;
    isSubscribed = true;//默认发送
    //常规配置订阅关系判断
    Set<SubItem> subItems = baseSubscriptionConfig.getItemSet();
    Iterator it = subItems.iterator();
    while (it.hasNext()) {
      SubItem subItem = (SubItem) it.next();
      String itemName = subItem.getName();
      String itemValue = subItem.getValue();
      String itemAction = subItem.getAction();
      if (!checkItemSubscripted(itemName, itemValue, itemAction, itemMap)) {
        isSubscribed = false;
        break;
      }
    }
    return isSubscribed;
  }


  private boolean isOtherSubscrpted(Map.Entry<String, String> entry, Map<String, Object> itemMap)
      throws Exception {
    boolean subscripted = false;
    if (!Strings.isNullOrEmpty(entry.getValue())) {
      //调用对应的订阅关系处理类
      String handleClass = entry.getValue();
      Class subscriptionClass = Class.forName(handleClass);
      Object obj = subscriptionClass.newInstance();
      Method method = subscriptionClass.getMethod("subscription", Map.class);
      subscripted = (boolean) method.invoke(obj, itemMap);
    } else {
      subscripted = false;
    }
    return subscripted;
  }

  /**
   * 检查是否满足选项配置要求
   *
   * @param itemName 选项名称
   * @param itemValue 选项值
   * @return 是否满足
   */
  private boolean checkItemSubscripted(String itemName, String itemValue, String itemAction,
      Map<String, Object> itemMap) {
    boolean subscripted = false;
    itemName = itemName.replaceAll("_", "");
    String value;
    if (itemNames.contains(itemName)) {
      Object obj = itemMap.get(itemName);
      if (obj instanceof String || obj instanceof Long || obj instanceof Integer) {
        value = String.valueOf(obj);
        String[] items = value.split(",");
        for (String item : items) {

          if ("0".equals(itemAction)) {
            if (itemValue.equals(item.split(":")[0])) {
              subscripted = true;
              break;
            }
          } else {
            String val = itemValue + ":" + itemAction;
            if (val.equals(item)) {
              subscripted = true;
              break;
            }
          }

        }

      }
//            else if (obj instanceof List) {
//                StringBuilder sb = new StringBuilder();
//                for (Object object : (List) obj) {
//                    sb.delete(0,sb.length());//清空
//                    value = sb.append(object).toString();
//                    if (itemValue.equals(value)) {
//                        subscripted = true;
//                        break;
//                    }
//                }
//            }
    }
    return subscripted;
  }

  /**
   * 将java bean转化成Map
   *
   * @param obj 对象
   * @return Map
   */
  private Map<String/*属性名称*/, Object/*属性值*/> convertObject2Map(Object obj) {
    if (obj == null) {
      return null;
    }
    Map<String, Object> itemMap = new HashMap<String, Object>();
    try {
      BeanInfo beanInfo = Introspector.getBeanInfo(obj.getClass());
      PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
      for (PropertyDescriptor property : propertyDescriptors) {
        String key = property.getName().toUpperCase();
        if (key.endsWith("S")) {//去除"S"
          key = key.substring(0, key.length() - 1);
        }

        // 过滤class属性
        if (!key.equals("class")) {
          // 得到property对应的getter方法
          Method getter = property.getReadMethod();
          Object value = getter.invoke(obj);
          itemMap.put(key, value);
        }

      }
    } catch (Exception e) {
      log.info("convertObject2Map Error " + e);
    }

    return itemMap;
  }
}
